package top.codedance.iotp.client.plugins.async.push;

import cd.io.netty.channel.ChannelHandlerContext;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import top.codedance.iotp.client.ClientApplication;
import top.codedance.iotp.common.executer.AsyncAbstractExecuter;
import top.codedance.iotp.client.plugins.async.com.ComExecuterImpl;
import top.codedance.iotp.common.constent.Command;
import top.codedance.iotp.common.entity.CommonEntity;
import org.apache.log4j.Logger;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class PushExecuterImpl extends AsyncAbstractExecuter {
    private static final Logger logger = ClientApplication.getLogger(PushExecuterImpl.class);
    private static ExecutorService pools = Executors.newCachedThreadPool();
    private static volatile Map<Integer, Set<String>> clients = new HashMap<>();
    private static volatile Queue<TaskEntity> tasks = new LinkedBlockingQueue<>();
    private static volatile CountDownLatch latch = new CountDownLatch(1);

    public PushExecuterImpl() {
        pools.submit(() -> {
            while (true) {
                try {
                    latch.await();
                    TaskEntity taskEntity = tasks.poll();
                    if (taskEntity != null) {
                        Set<Map.Entry<Integer, Set<String>>> entries = clients.entrySet();
                        for (Map.Entry<Integer, Set<String>> entry : entries) {
                            if (taskEntity.getInstance() == entry.getKey()) {
                                for (String cid : entry.getValue()) {
                                    CommonEntity ce = genCommonEntity();
                                    JSONObject jo = new JSONObject();
                                    jo.put("id", taskEntity.getInstance());
                                    jo.put("data", taskEntity.getData());
                                    ce.setBody(jo.toJSONString().getBytes(StandardCharsets.UTF_8));
                                    send(ce.build(), cid);
                                }
                            }
                        }
                    }
                }catch (Exception e){
                    logger.debug(e.getMessage());
                }finally {
                    latch = new CountDownLatch(1);
                }
            }
        });
    }

    @Override
    public void exec(ChannelHandlerContext channelHandlerContext, CommonEntity commonEntity, String taskId) {
        super.exec(channelHandlerContext, commonEntity, taskId);
        pools.submit(() -> {
            JSONObject requestBodyJSONObject = getRequestBodyJSONObject();
            if(requestBodyJSONObject.containsKey("body")){
                JSONArray body = requestBodyJSONObject.getJSONArray("body");
                for (int i = 0;i < body.size();i++){
                    Integer instance = body.getInteger(i);
                    if(!clients.containsKey(instance)){
                        clients.put(instance, new LinkedHashSet<>());
                    }
                    clients.get(instance).add(getClientId());
                }
                Integer instance = getInstance();
                if(!clients.containsKey(instance)){
                    clients.put(instance, new LinkedHashSet<>());
                }
                clients.get(instance).add(getClientId());
            }
            if(requestBodyJSONObject.containsKey("release")){
                synchronized (ComExecuterImpl.class) {
                    Set<Map.Entry<Integer, Set<String>>> entries = clients.entrySet();
                    for (Map.Entry<Integer, Set<String>> entry : entries) {
                        try {
                            entry.getValue().remove(getClientId());
                        } catch (Exception e) {
                        }
                    }
                }
            }
        });
    }

    public static void triger(Integer instance, byte[] data){
        synchronized (PushExecuterImpl.class){
            tasks.add(new TaskEntity(instance, new String(data, StandardCharsets.UTF_8)));
            latch.countDown();
        }
    }

    /**
     * 统一获取发送实体
     * @return
     */
    private CommonEntity genCommonEntity(){
        CommonEntity ce = new CommonEntity();
        ce.setCommand(Command.ASYNC_PUSH_DATA_COMMAND_RECEIVE);
        ce.setVersion(ClientApplication.getDataVersion());
        return ce;
    }
}
